#include "../RpcChannel.h"
#include <google/protobuf/descriptor.h>

using namespace yhaida;
using namespace yhaida::rpc;

RpcChannel::RpcChannel()
    : _codec(std::bind(&RpcChannel::onRpcMessage, this, _1, _2, _3)),
      _services(NULL)
{
    LOG_INFO << "RpcChannel::ctor - " << this;
}

RpcChannel::RpcChannel(const TcpConnectionPtr &conn)
    : _codec(std::bind(&RpcChannel::onRpcMessage, this, _1, _2, _3)),
      _conn(conn),
      _services(NULL)
{
    LOG_INFO << "RpcChannel::ctor - " << this;
}

RpcChannel::~RpcChannel()
{
    LOG_INFO << "RpcChannel::dtor - " << this;
    for (const auto &item : _outstandings)
    {
        OutstandingCall outstanding = item.second;
        delete outstanding.response;
        delete outstanding.done;
    }
}
/**
 * @brief  客户端调用：创建message->序列化->send
 */
void RpcChannel::CallMethod(const ::google::protobuf::MethodDescriptor *method,
                            ::google::protobuf::RpcController *controller,
                            const ::google::protobuf::Message *request,
                            ::google::protobuf::Message *response,
                            ::google::protobuf::Closure *done)
{
    LOG_DEBUG;
    RpcMessage message;
    message.set_type(MessageType::REQUEST);
    int64_t id = _id.incrementAndGet();
    message.set_id(id);
    /**
     * @brief 相同的返回值std::string
     * method->service()->full_name()
     * service->GetDescriptor()->full_name()
     * 
     */
    message.set_service(method->service()->full_name());
    message.set_method(method->name());
    message.set_request(request->SerializeAsString());//序列化

    //根据ID将response保存起来，等待responce
    OutstandingCall outstanding = {response, done};
    {
        MutexLockGuard guard(_mutex);
        _outstandings[id] = outstanding;
    }

    //发送数据
    _codec.send(_conn, message);
    LOG_DEBUG;
}

/**
 * @brief   服务器端调用
 *
 */
void RpcChannel::onMessage(const TcpConnectionPtr &conn,
                           Buffer *buffer,
                           Timestamp receiveTime)
{
    _codec.onMessage(conn, buffer, receiveTime);
}
/**
 * @brief   _codec 解码的时候
 */
void RpcChannel::onRpcMessage(const TcpConnectionPtr &conn,
                              const RpcMessagePtr &messagePtr,
                              Timestamp receiveTime)
{
    assert(conn == _conn);
    RpcMessage &message = *messagePtr;
    if (message.type() == MessageType::RESPONSE)
    {
        int64_t id = message.id();

        assert(message.has_response() || message.has_error());

        OutstandingCall outstanding = {NULL, NULL};
        {
            MutexLockGuard guard(_mutex);
            auto it = _outstandings.find(id);
            if (it != _outstandings.end())
            {
                outstanding = it->second;
                _outstandings.erase(it);
            }
        }
        if (outstanding.response)
        {
            std::unique_ptr<google::protobuf::Message> d(outstanding.response);
            if (message.has_response())
            {
                outstanding.response->ParseFromString(message.response());
            }
            if (outstanding.done)
            {
                outstanding.done->Run();//同步
            }
        }
    }
    else if (message.type() == MessageType::REQUEST)
    {
        LOG_DEBUG;
        ErrorCode error = ErrorCode::WRONG_PROTO;
        if (_services)
        {
            auto it = _services->find(message.service());
            if (it != _services->end())
            {
                LOG_DEBUG<<"service: "<<message.service();
                LOG_DEBUG<<"method: "<<message.method();
                google::protobuf::Service *service = it->second;
                assert(service != NULL);
                const google::protobuf::ServiceDescriptor *desc = service->GetDescriptor();
                const google::protobuf::MethodDescriptor *method = desc->FindMethodByName(message.method());
                /**
                 * @brief  
                 * method[0]   rpc Solve (SudokuRequest) returns (SudokuResponse);
                 * 
                 */
                if (method)
                {
                    LOG_DEBUG;
                    std::unique_ptr<google::protobuf::Message> request(service->GetRequestPrototype(method).New());//如：SudokuRequest
                    //拿到客户端发来的message
                    if(request->ParseFromString(message.request()))//反序列化
                    {
                        google::protobuf::Message* response = service->GetResponsePrototype(method).New();//如：SudokuResponse
                        int64_t id = message.id();
                        /**
                         * @brief   CallMethod()调用Solve()服务方法了
                         * 
                         */
                        service->CallMethod(method,NULL,get_pointer(request),response,
                                    NewCallback(this,&RpcChannel::doneCallback,response,id));
                        error = ErrorCode::NO_ERROR;
                    }
                    else
                    {
                        error = ErrorCode::INVALID_REQUEST;
                    }
                }
                else
                {

                    error = ErrorCode::NO_METHOD; 
                }
            }
            else
            {
                error = ErrorCode::NO_SERVICE;
            }
        }
        else
        {
            error = ErrorCode::NO_SERVICE;
        }
        if (error != ErrorCode::NO_ERROR)//未出现错误
        {
            RpcMessage response;
            response.set_type(MessageType::RESPONSE);
            response.set_id(message.id());
            response.set_error(error);
            _codec.send(_conn, response);
        }
    }
    else if (message.type() == MessageType::ERROR)
    {
    }
    LOG_DEBUG;
}
void RpcChannel::doneCallback(::google::protobuf::Message *response, int64_t id)
{
    std::unique_ptr<google::protobuf::Message> d(response);
    RpcMessage message;
    message.set_type(MessageType::RESPONSE);
    message.set_id(id);
    message.set_response(response->SerializeAsString());
    _codec.send(_conn, message);
}